package com.ysw.rpc.client.component;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.nacos.api.naming.listener.NamingEvent;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.ysw.rpc.client.config.ClientProperties;
import com.ysw.rpc.client.enums.ConnectionEnum;
import com.ysw.rpc.client.handler.initializer.ClientInitializerHandler;
import com.ysw.rpc.common.constants.RpcConstants;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOutboundInvoker;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;

import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static com.ysw.rpc.client.utils.CacheUtil.*;

/**
 * @author YinShangwen
 * @since 2023/4/18 13:10
 */
@Slf4j
public class ChannelComponent {
    private final ClientProperties clientProperties;
    private final NacosComponent nacosComponent;
    private final Bootstrap bootstrap;
    private final NioEventLoopGroup eventLoopGroup;

    public ChannelComponent(ClientProperties clientProperties, NacosComponent nacosComponent) {
        this.clientProperties = clientProperties;
        this.nacosComponent = nacosComponent;
        this.eventLoopGroup = new NioEventLoopGroup(new DefaultThreadFactory("y-rpc-client", true));
        this.bootstrap = new Bootstrap()
                .group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ClientInitializerHandler(clientProperties));
    }

    /**
     * 获取一个channel
     *
     * @param serviceName serviceName
     * @return channel
     */
    public Channel getChannel(String serviceName) {
        if (ConnectionEnum.isSingleLong(clientProperties.getConnectionType())) {
            Map<String, Channel> channelMap = getServerChannelCache().get(serviceName);
            AtomicInteger count = getServerCountCache().get(serviceName);
            int idx = count.getAndIncrement() % Math.max(channelMap.values().size(), 1);
            for (Channel channel : channelMap.values()) {
                if (idx-- == 0) {
                    return channel;
                }
            }
            return null;
        }
        if (ConnectionEnum.isShort(clientProperties.getConnectionType())) {
            Instance instance = nacosComponent.selectOneHealthyInstance(serviceName);
            if (Objects.isNull(instance)) {
                if (log.isInfoEnabled()) {
                    log.info("subscribe error: no subscribe, service: {}", serviceName);
                }
                return null;
            }
            try {
                JSONObject object = JSON.parseObject(instance.getMetadata().get(RpcConstants.METADATA_KEY));
                return bootstrap.connect(new InetSocketAddress(instance.getIp(), object.getInteger(RpcConstants.METADATA_PORT_KEY))).sync().channel();
            } catch (InterruptedException e) {
                return null;
            }
        }
        return null;
    }

    /**
     * 初始化channel
     */
    public void init() {
        if (clientProperties.getSubscribes().isEmpty()) {
            // subscribe all
            nacosComponent.getServices().forEach(this::initHandler);
        } else {
            clientProperties.getSubscribes().forEach(this::initHandler);
        }
        if (log.isInfoEnabled()) {
            log.info("client init ok...");
        }
    }

    /**
     * 初始化处理
     *
     * @param serviceName 服务名
     */
    private void initHandler(String serviceName) {
        Instance instance = nacosComponent.selectOneHealthyInstance(serviceName);
        if (Objects.isNull(instance)) {
            if (log.isInfoEnabled()) {
                log.info("nacos error: not healthy instance, service: {}", serviceName);
            }
            return;
        }

        if (!instance.getMetadata().containsKey(RpcConstants.METADATA_KEY)) {
            // 不是rpc服务提供方
            return;
        }
        // 处理接口与服务对应关系
        JSONObject jsonObject = JSON.parseObject(instance.getMetadata().get(RpcConstants.METADATA_KEY));
        for (Object clazz : jsonObject.getJSONArray(RpcConstants.METADATA_API_KEY)) {
            getClassServerCache().putIfAbsent(clazz.toString(), serviceName);
        }

        subscribe(serviceName, instance);
    }

    /**
     * 订阅
     *
     * @param serviceName serviceName
     */
    private void subscribe(String serviceName, Instance instance) {
        if (!ConnectionEnum.isSingleLong(clientProperties.getConnectionType()) || getServerChannelCache().containsKey(serviceName)) {
            // 非单一长连接 或 已订阅
            return;
        }
        getServerChannelCache().putIfAbsent(serviceName, new ConcurrentHashMap<>(16));
        getServerCountCache().put(serviceName, new AtomicInteger(1));

        connect(serviceName, instance);

        nacosComponent.subscribe(serviceName, event -> {
            NamingEvent namingEvent = (NamingEvent) event;
            List<Instance> instances = namingEvent.getInstances();
            disconnect(serviceName, instances);
            instances.forEach(item -> connect(serviceName, item));
        });
    }

    /**
     * 剔除失活服务
     *
     * @param serviceName serviceName
     * @param instances   instances
     */
    private void disconnect(String serviceName, List<Instance> instances) {
        Map<String, Channel> channelMap = getServerChannelCache().get(serviceName);
        List<String> instanceIds = instances.stream().map(Instance::getInstanceId).collect(Collectors.toList());
        Collection<String> diffInstanceIds = CollectionUtils.subtract(channelMap.keySet(), instanceIds);
        for (String diffInstanceId : diffInstanceIds) {
            Channel channel = channelMap.remove(diffInstanceId);
            if (Objects.nonNull(channel)) {
                channel.close();
            }
        }
    }

    /**
     * 连接服务
     *
     * @param serviceName serviceName
     * @param instance    instance
     */
    private void connect(String serviceName, Instance instance) {
        Map<String, Channel> channelMap = getServerChannelCache().get(serviceName);
        String instanceId = String.join("_", instance.getIp(), String.valueOf(instance.getPort()));
        if (channelMap.containsKey(instanceId)) {
            // 已有连接
            return;
        }
        JSONObject object = JSON.parseObject(instance.getMetadata().get(RpcConstants.METADATA_KEY));
        ChannelFuture channelFuture = bootstrap.connect(new InetSocketAddress(instance.getIp(), object.getInteger(RpcConstants.METADATA_PORT_KEY)));
        channelFuture.addListener(future -> {
            if (future.isSuccess()) {
                channelMap.put(instanceId, channelFuture.channel());
            }
        });
    }

    /**
     * 取消订阅一个服务
     *
     * @param serviceName serviceName
     */
    public void unsubscribe(String serviceName) {
        nacosComponent.unsubscribe(serviceName);
        getServerChannelCache().get(serviceName).values().forEach(ChannelOutboundInvoker::close);
    }

    public NioEventLoopGroup getEventLoopGroup() {
        return eventLoopGroup;
    }
}
